package com.hmall.search.autotoLoad;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.hmall.search.constant.ItemConstant;
import com.hmall.search.mapper.SearchMapper;
import com.hmall.search.pojo.Item;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @Auth: wan
 * @Desc: ES自动数据插入
 * @Date: 2023/4/19 下午 10:06
 */
@Component
@Slf4j
public class MyApplicationRunner implements ApplicationRunner {

	@Autowired
	private SearchMapper searchMapper;

	@Autowired
	private RestHighLevelClient restHighLevelClient;

    @Override
    public void run(ApplicationArguments args) throws Exception {
	    QueryWrapper<Item> queryWrapper = new QueryWrapper<>();
	    queryWrapper.eq(ItemConstant.ITEM_STATUS,ItemConstant.ITEM_STATUS_NUM);
	    List<Item> items = searchMapper.selectList(queryWrapper);

		BatchInsertion(items,restHighLevelClient);
	    log.info("数据批量添加成功------------------------->Succeeded in adding data in batches. Procedure");
    }

	/**
	 * 功能描述:
	 * @param data 需要插入的数据
	 * @param client 客户端对象
	 * @return : void
	 */
	public static void BatchInsertion(List<Item> data, RestHighLevelClient client) throws IOException, InterruptedException {

		// 创建线程
		List<Thread> threads = new ArrayList<>();
		// 获取每个线程分配的数据条数
		int batchSize = data.size() / ItemConstant.THREAD_COUNT;
		for (int i = 0; i < ItemConstant.THREAD_COUNT; i++) {
			// 线程执行的条数
			int start = i * batchSize;

			int end = (i + 1) * batchSize;
			if (i == ItemConstant.THREAD_COUNT - 1) {
				end = data.size();
			}
			List<Item> subData = data.subList(start, end);
			ElasticsearchInsertThread thread = new ElasticsearchInsertThread(subData, client);
			threads.add(thread);
		}

		// 启动线程
		for (Thread thread : threads) {
			thread.start();
		}

		// 等待线程完成
		for (Thread thread : threads) {
			try {
				thread.join();
			} catch (InterruptedException e) {
				// handle exceptions
				throw new InterruptedException();
			}
		}
	}
}